﻿///////////////////////////////////////////////////////////////////////////////
// Copyright 2016 lingzimaike@gmail.com
// GPLv2
// Author: lingzimaike
///////////////////////////////////////////////////////////////////////////////

#include "papa/net/tcp_server.h"
#include <memory.h>
#include <cassert>
#include "papa/net/socket_event_handler.h"
#include "papa/net/socket_event_handler_factory.h"
#include "papa/net/io_runnable.h"
#include "papa/net/tcp_socket.h"

namespace papa {
namespace net {

	TcpServer::TcpServer(SocketEventHandlerFactory* factory)
		: state_(NOT_READY)
		, conn_id_gen_(0)
		, acceptor_(NULL)
		, event_handler_factory_(factory)
		, thread_array_()
		, socket_map_()
		, socket_map_lock_()
	{
		assert(factory);
	}

	TcpServer::~TcpServer(void)
	{

	}

	bool TcpServer::Open(const ::std::string& host, int port, 
		int backlog /* = 1024 */, int num_thread /* = 1 */)
	{
		assert(state_ == NOT_READY);
		bool exit_value = false;

		do
		{
			thread::Thread* default_thread = init_io_threads(num_thread);
			if (!default_thread) {
				break;
			}
			IORunnable* default_runnable = static_cast<IORunnable*>(default_thread->get_runnable());

			struct sockaddr_in saddr;
			memset(&saddr, 0, sizeof(saddr));
			saddr.sin_family = AF_INET;
#if defined(__WINDOWS__)
			saddr.sin_addr.S_un.S_addr = ::inet_addr(host.c_str());
#elif defined(__LINUX__)
			saddr.sin_addr.s_addr = ::inet_addr(addr.c_str());
#endif
			saddr.sin_port = ::htons(u_short(port));

			acceptor_ = evconnlistener_new_bind(
				default_runnable->event_base_, listener_cb, this,
				LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE, backlog,
				(struct sockaddr*)&saddr,
				sizeof(saddr));
			if (!acceptor_) break;

			if (!start_io_threads()) {
				break;
			}

			state_ = RUNNING;
			exit_value = true;
		} while (false);

		if (!exit_value)
		{
			for (ThreadArray::size_type x = 0;
				x < thread_array_.size(); ++x)
			{
				static_cast<IORunnable*>(thread_array_[x]->get_runnable())->Stop();
				thread_array_[x]->Join();
			}
			thread_array_.clear();

			if (acceptor_)
			{
				evconnlistener_free(acceptor_);
				acceptor_ = NULL;
			}
		}
		return exit_value;
	}

	void TcpServer::Join(void)
	{
		assert(state_ == RUNNING || state_ == EXITING);
		for (ThreadArray::size_type x = 0;
			x < thread_array_.size(); ++x)
		{
			thread_array_[x]->Join();
		}
		thread_array_.clear();
		state_ = EXIT;
	}

	void TcpServer::Close(void)
	{
		assert(state_ == RUNNING);
		state_ = EXITING;

		if (acceptor_)
		{
			evconnlistener_free(acceptor_);
			acceptor_ = NULL;
		}

		{
			PAPA_LOCK_MUTEX(socket_map_lock_);
			for (SocketMap::iterator it = socket_map_.begin();
				it != socket_map_.end(); )
			{
				it->second->do_close();
				it->second->event_handler_->OnClose();

                TcpSocket* tcp_socket = it->second;
                it = socket_map_.erase(it);
                delete tcp_socket;
			}
			socket_map_.clear();
		}

		for (ThreadArray::size_type x = 0;
			x < thread_array_.size(); ++x) {
				static_cast<IORunnable*>(thread_array_[x]->get_runnable())->Stop();
		}
	}

	void TcpServer::do_close_socket(TcpSocket* sock)
	{
		assert(sock);
		sock->actor_->DecrementLoad();
		PAPA_LOCK_MUTEX(socket_map_lock_);
		SocketMap::iterator it = socket_map_.find(sock->id_);
		if (it != socket_map_.end()) {
            TcpSocket* tcp_socket = it->second;
			socket_map_.erase(it);
            delete tcp_socket;
		}
	}

	thread::Thread* TcpServer::find_most_free_thread(void) const
	{
		ThreadArray::size_type idx;
		int minLoad = INT_MAX;
		for (ThreadArray::size_type x = 0;
			x < thread_array_.size(); ++x)
		{
			int load = static_cast<IORunnable*>(thread_array_[x]->get_runnable())->get_load();
			if (load < minLoad)
			{
				minLoad = load;
				idx = x;
			}
		}
		return thread_array_[idx].get();
	}

	thread::Thread* TcpServer::init_io_threads(int num_thread)
	{
        bool init_ok = true;
        for (int cur_thread = 0; cur_thread < num_thread; ++cur_thread)
        {
            IORunnable* runnable = new IORunnable();
            if ((runnable == NULL) || !runnable->Open()) {
                init_ok = false;
                break;
            }

            ::boost::shared_ptr<thread::Thread> thread(new thread::Thread(runnable));
            if (!thread) {
                delete runnable;
                init_ok = false;
                break;
            }

            thread_array_.push_back(thread);
        }

        if (!init_ok) {
            thread_array_.clear();
        }
        return (init_ok ? thread_array_[0].get() : NULL);
	}

	bool TcpServer::start_io_threads(void)
	{
		for (ThreadArray::size_type x = 0;
			x < thread_array_.size(); ++x)
		{
			if (!thread_array_[x]->Start()) {
				return false;
			}
		}
		return true;
	}

	void TcpServer::listener_cb(struct evconnlistener *listener, evutil_socket_t fd, 
		struct sockaddr *sa, int socklen, void *user_data)
	{
		TcpServer* tcp_server_ptr = static_cast<TcpServer*>(user_data);
		assert(tcp_server_ptr);
		tcp_server_ptr->OnAccept(fd, sa, socklen);
	}

	void TcpServer::OnAccept(int fd, struct sockaddr *sa, int socklen)
	{
		do 
		{
			int new_conn_id = conn_id_gen_ + 1;
			{
				PAPA_LOCK_MUTEX(socket_map_lock_);
				if (socket_map_.find(new_conn_id) != socket_map_.end()) {
					break;
				}
				++conn_id_gen_;
			}

			::boost::scoped_ptr<SocketEventHandler> event_handler(event_handler_factory_->Create());
			if (!event_handler) {
				break;
			}
			
			thread::Thread* free_thread = find_most_free_thread();
			IORunnable* free_runnable = static_cast<IORunnable*>(free_thread->get_runnable());

            TcpSocket* tcp_socket = new TcpSocket(event_handler);
			if (tcp_socket != NULL)
			{
				if (tcp_socket->Open(fd, free_runnable->event_base_, sa, socklen))
				{
					tcp_socket->id_ = new_conn_id;
					tcp_socket->actor_ = free_runnable;
					tcp_socket->owner_ = this;
					{
						PAPA_LOCK_MUTEX(socket_map_lock_);
                        socket_map_[new_conn_id] = tcp_socket;
					}

					tcp_socket->event_handler_->OnOpen();
                    return;
				}
			}

		} while (false);

		TcpSocket::CloseSocket(fd);
	}

}}  // namespace papa::net